package com.gitee.xhs.savePointReader.keyedState.aggregatingStateReader;

import com.gitee.xhs.keyedState.aggregatingState.AggregateFunctionDao;
import com.gitee.xhs.keyedState.aggregatingState.AggregatingStateConstant;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/6/30 15:23
 * @description AggregatingStateReaderFunction
 */
public class AggregatingStateReaderFunction extends KeyedStateReaderFunction<Long, AggregatingStateExhibitor> {

    private final String jobId;
    private final String operatorUid;
    private transient AggregatingState<Transaction, AggregateFunctionDao> aggregatingState;

    public AggregatingStateReaderFunction(String jobId, String operatorUid) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        aggregatingState = getRuntimeContext().getAggregatingState(AggregatingStateConstant.AGGREGATING_STATE_DESCRIPTOR);
    }

    @Override
    public void readKey(Long key, Context ctx, Collector<AggregatingStateExhibitor> out) throws Exception {
        AggregatingStateExhibitor aggregatingStateExhibitor = new AggregatingStateExhibitor();
        aggregatingStateExhibitor.setJobId(jobId);
        aggregatingStateExhibitor.setOperatorUid(operatorUid);
        aggregatingStateExhibitor.setKey(key);
        aggregatingStateExhibitor.setAggregatingStateValue(aggregatingState.get().getAmountUniqueCountMap());
        out.collect(aggregatingStateExhibitor);
    }
}
